package com.atguigu.chapter11.window;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink05_Window_TVF_group_set {
    public static void main(String[] args) {

        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((ws,ts)->ws.getTs())
                );

        // 获取表执行环境，在这之前要获取流的执行环境作为参数传过来
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 把流转换成表
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"), $("et").rowtime());
        tEnv.createTemporaryView("sensor",table);

        /*tEnv.sqlQuery("select" +
                " window_start,window_end,id," +
                " sum(vc) vc_sum " +
                " from table(tumble( table sensor,descriptor(et),interval '5' second)) " +
                " group by window_start,window_end,id " +
                " union " +
                "select window_start,window_end,'a' id," +
                " sum(vc) vc_sum " +
                " from table(tumble( table sensor,descriptor(et),interval '5' second)) " +
                " group by window_start,window_end ")
                .execute().print();*/

        tEnv
                .sqlQuery("select " +
                        " window_start,window_end,id," +
                        " sum(vc) vc_sum " +
                        " from table(tumble(table sensor,descriptor(et),interval '5' second)) " +
//                        " group by window_start,window_end,grouping sets((id),())")    // 这种就是有id和没id
//                        " group by window_start,window_end, rollup(id)")    // 还有这种写法
                        " group by window_start,window_end, cube(id)")    // 构建cube，多维立方体
                .execute().print();
    }
}
